package me.guowenlong.transformer.core;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Completable;
import io.reactivex.CompletableSource;
import io.reactivex.CompletableTransformer;
import io.reactivex.Flowable;
import io.reactivex.FlowableTransformer;
import io.reactivex.Maybe;
import io.reactivex.MaybeSource;
import io.reactivex.MaybeTransformer;
import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.Single;
import io.reactivex.SingleSource;
import io.reactivex.SingleTransformer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;
import me.guowenlong.transformer.bean.RetryBean;
import me.guowenlong.transformer.compoment.FlowRetryFunction;
import me.guowenlong.transformer.compoment.ObserveRetryFunction;

/**
 * desc   : 扩展Transformer
 * author : guowenlong
 * Email  : guowenlong20000@gmail.com
 * time   : 2019年01月02日15 : 18
 */
public class ExtendTransformer<T> implements ObservableTransformer<T, T>, FlowableTransformer<T, T>, SingleTransformer<T, T>, MaybeTransformer<T, T>, CompletableTransformer {

    private Function<T, Observable<T>> flapMap = Observable::just;
    private Function<Throwable, Observable<T>> onErrorResumeNext = Observable::error;
    private Function<Throwable, RetryBean> retryWhen = (Function<Throwable, RetryBean>) throwable -> RetryBean.getNilRetryBean();
    private Consumer<Throwable> onErrorConsumer = throwable -> {};

    public ExtendTransformer() {}

    public void setFlapMap(Function<T, Observable<T>> flapMap) {
        this.flapMap = flapMap;
    }

    public void setOnErrorResumeNext(Function<Throwable, Observable<T>> onErrorResumeNext) {
        this.onErrorResumeNext = onErrorResumeNext;
    }

    public void setRetryWhen(Function<Throwable, RetryBean> retryWhen) {
        this.retryWhen = retryWhen;
    }

    public void setOnErrorConsumer(Consumer<Throwable> onErrorConsumer) {
        this.onErrorConsumer = onErrorConsumer;
    }

    @Override
    public Observable<T> apply(Observable<T> upstream) {
        return upstream
                .flatMap((Function<T, ObservableSource<T>>) t -> flapMap.apply(t))
                .observeOn(AndroidSchedulers.mainThread())
                .onErrorResumeNext((Function<Throwable, ObservableSource<T>>) throwable -> onErrorResumeNext.apply(throwable))
                .retryWhen(new ObserveRetryFunction(retryWhen))
                .doOnError(onErrorConsumer)
                .subscribeOn(Schedulers.io());
    }

    @Override
    public Flowable<T> apply(Flowable<T> upstream) {
        return upstream
                .flatMap((Function<T, Flowable<T>>) t -> flapMap.apply(t).toFlowable(BackpressureStrategy.BUFFER))
                .observeOn(AndroidSchedulers.mainThread())
                .retryWhen(new FlowRetryFunction(retryWhen))
                .doOnError(onErrorConsumer)
                .subscribeOn(Schedulers.io());
    }

    @Override
    public CompletableSource apply(Completable upstream) {
        return upstream
                .onErrorResumeNext(throwable -> onErrorResumeNext.apply(throwable).ignoreElements())
                .observeOn(AndroidSchedulers.mainThread())
                .retryWhen(new FlowRetryFunction(retryWhen))
                .doOnError(onErrorConsumer)
                .subscribeOn(Schedulers.io());
    }

    @Override
    public MaybeSource<T> apply(Maybe<T> upstream) {
        return upstream
                .flatMap((Function<T, MaybeSource<T>>) t -> flapMap.apply(t).firstElement())
                .observeOn(AndroidSchedulers.mainThread())
                .onErrorResumeNext(throwable -> {
                    return onErrorResumeNext.apply(throwable).firstElement();
                })
                .retryWhen(new FlowRetryFunction(retryWhen))
                .doOnError(onErrorConsumer)
                .subscribeOn(Schedulers.io());
    }

    @Override
    public SingleSource<T> apply(Single<T> upstream) {
        return upstream
                .flatMap((Function<T, SingleSource<T>>) t -> flapMap.apply(t).firstOrError())
                .observeOn(AndroidSchedulers.mainThread())
                .onErrorResumeNext(throwable -> onErrorResumeNext.apply(throwable).firstOrError())
                .retryWhen(new FlowRetryFunction(retryWhen))
                .doOnError(onErrorConsumer)
                .subscribeOn(Schedulers.io());
    }
}